package simple;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Random;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

public class HolmesKafkaProducer {

  private static ObjectMapper objectMapper = new ObjectMapper();
  private static String topic = "holmes-api-default-holmes-call-log";

  public static void main(String[] args) throws IOException {
    String brokers = "10.57.16.13:9092";

    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringSerializer");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringSerializer");
    KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

    List<Map<String, String>> list = new ArrayList<>();
    list.add(new HashMap<>());
    list.add(new HashMap<>());

    list.get(0).put("modelUuid", "2022082685779634");
    list.get(0).put("modelName", "liux_lightGBM");
    list.get(0).put("modelUuid", "2022072652237425");
    list.get(0).put("modelName", "评分卡001");

    LocalDateTime start = LocalDateTime.now();
    LocalDateTime end = start.plusMinutes(5);
    int cnt = 0;
    while (true) {
      for (int i = 0; i < 10; i++) {
        ++cnt;
        String str = buildLog(i, list);
        ProducerRecord message = new ProducerRecord<String, String>(topic,
            String.valueOf(new Random().nextInt(3)), str);
        producer.send(message);
        System.out.println("write msg " + cnt);
      }
      if (LocalDateTime.now().isAfter(end)) {
        System.out.println(("time is end"));
        return;
      }
    }
  }

  private static String buildLog(int i, List<Map<String, String>> list) throws IOException {
    int idx = i % list.size();
    boolean odd = new Random().nextInt(10) % 2 == 1;
    Map<String, String> m = list.get(idx);

    String temp = "{\"modelEmptyInputs\":[\"age\",\"sex\",\"RevolvingUtilizationOfUnsecuredLines\",\"NumberOfTime30-59DaysPastDueNotWorse\",\"DebtRatio\",\"MonthlyIncome\",\"NumberOfOpenCreditLinesAndLoans\",\"NumberOfTimes90DaysLate\"],\"inputData\":{},\"data\":{\"probability\":[0.7103809797849969,0.289619020215003],\"target\":0},\"batchData\":[],\"inputs\":{\"NumberOfOpenCreditLinesAndLoans\":-999,\"NumberOfTime30-59DaysPastDueNotWorse\":-999,\"RevolvingUtilizationOfUnsecuredLines\":-999,\"DebtRatio\":-999,\"NumberOfTimes90DaysLate\":-999,\"sex\":-999,\"age\":-999,\"MonthlyIncome\":-999},\"modelVersion\":\"V1\",\"diffDatas\":{},\"index\":\"2022082685779634V1\",\"modelType\":\"P\",\"modelUuid\":\"2022061522531913\",\"modelName\":\"liux_lightGBM\",\"callNumber\":\"5fe9014ce40846068bf32ca7b8647244\",\"pmmlInput\":{},\"success\":false,\"testGroup\":false,\"id\":\"1661755219786-90866742\",\"seqId\":\"2022082975782697\",\"modelTestFlag\":\"holmes_web_model_test_flag_@tongdun\",\"spendTime\":130,\"timestamp\":1661755219634}";
    Map<String, Object> map = objectMapper.readValue(temp,
        new TypeReference<Map<String, Object>>() {
        });

    String timestamp = DateTimeFormatter.ofPattern("yyyyMMddHHmmssSSS")
        .format(LocalDateTime.now());

    map.put("modelUuid", m.get("modelUuid"));
    map.put("modelName", m.get("modelName"));
    map.put("spendTime", new Random().nextInt(100));
    map.put("timestamp", System.currentTimeMillis());
    map.put("seqId", timestamp);
    map.put("success", odd);
    return objectMapper.writeValueAsString(map);
  }
}
